package ext

import (
	"fmt"
	"github.com/icetech233/ekuiper-kafka/conf"
	"strings"

	kafkago "github.com/segmentio/kafka-go"
)

type kafkaSourceConf struct {
	Brokers     string `json:"brokers"`
	Topic       string `json:"topic"`
	GroupID     string `json:"groupID"`
	Partition   int    `json:"partition"`
	MaxAttempts int    `json:"maxAttempts"`
	MaxBytes    int    `json:"maxBytes"`
}

func (c *kafkaSourceConf) validate() error {
	if len(strings.Split(c.Brokers, ",")) == 0 {
		return fmt.Errorf("brokers can not be empty")
	}
	return nil
}

func (c *kafkaSourceConf) GetReaderConfig(topic string) kafkago.ReaderConfig {
	return kafkago.ReaderConfig{
		Brokers:     strings.Split(c.Brokers, ","),
		GroupID:     c.GroupID,
		Topic:       topic,
		Partition:   c.Partition,
		MaxBytes:    c.MaxBytes,
		MaxAttempts: c.MaxAttempts,
	}
}

func getSourceConf(props map[string]interface{}) (*kafkaSourceConf, error) {
	c := &kafkaSourceConf{
		MaxBytes:    1e6,
		MaxAttempts: 3,
	}
	err := conf.MapToStruct(props, c)
	if err != nil {
		return nil, fmt.Errorf("read properties %v fail with error: %v", props, err)
	}
	return c, nil
}
